package com.diy.sigmund.diyorder.common.middleware.mq.rocketmq.spring.one;

import com.diy.sigmund.diycommon.test.entity.MqTestVO;
import com.diy.sigmund.diycommon.util.json.JacksonUtil;
import java.util.Objects;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;


/**
 * <br>普通消费
 * <br>实现RocketMQListener接口
 * <br>申明泛型对象即业务对象
 * <br>使用@RocketMQMessageListener注解
 * <br>配置@RocketMQMessageListener注
 * <br>必须配置项入下
 * <br>group
 * <br>topic
 * <br>使用@Component注解或者子注解
 * <br>实现接口方法
 *
 * @author ylm-sigmund
 * @since 2022/5/5 22:21
 */
@Component
@RocketMQMessageListener(
        // 要求相同角色的消费者拥有完全相同的订阅和消费者组才能正确实现负载均衡。它是必需的，并且必须是全球唯一的。
        consumerGroup = "${group.consumer.sendDataType}",
        // %DLQ%sendDataType-consumer
        topic = "${topic.sendDataType.DLQ}",
        // 控制如何选择消息。也可以看看：SelectorType
        selectorType = SelectorType.TAG,
        // 控制可以选择的消息。语法请参见SelectorType.TAG和SelectorType.SQL92
        selectorExpression = "*",
        // 控制消费模式，可选择并发或有序接收消息。
        consumeMode = ConsumeMode.CONCURRENTLY,
        // 控制消息模式，如果希望所有订阅者都接收消息，广播是一个不错的选择。
        messageModel = MessageModel.CLUSTERING,
        // 消费者线程号。
        consumeThreadNumber = 20,
        // 最大重复消费次数。在并发模式下，-1 表示 16；在有序模式下，-1 表示 Integer.MAX_VALUE。
        maxReconsumeTimes = 2,
        // 一条消息可能阻塞消费线程的最长时间（以分钟为单位）
        consumeTimeout = 15L,
        // 发送回复消息超时。
        replyTimeout = 3000,
        // “访问密钥”的属性。
        accessKey = RocketMQMessageListener.ACCESS_KEY_PLACEHOLDER,
        // “密钥”的属性。
        secretKey = RocketMQMessageListener.SECRET_KEY_PLACEHOLDER,
        // 消息跟踪的切换标志实例。
        enableMsgTrace = false,
        // 消息跟踪主题的名称值。如果不配置，可以使用默认的跟踪主题名称。
        customizedTraceTopic = RocketMQMessageListener.TRACE_TOPIC_PLACEHOLDER,
        nameServer = RocketMQMessageListener.NAME_SERVER_PLACEHOLDER,
        // “访问通道”的属性。
        accessChannel = RocketMQMessageListener.ACCESS_CHANNEL_PLACEHOLDER,
        // “tlsEnable”的属性默认为false。
        tlsEnable = "false",
        // 消费者的命名空间。
        namespace = "",
        // 并发模式下的消息消费重试策略。 -1，不重试，直接放入DLQ ；0，broker控制重试频率；>0，client控制重试频率
        delayLevelWhenNextConsume = 2,
        // 在有序模式下暂停拉取的时间间隔，以毫秒为单位。最小值为 10，最大值为 30000。
        suspendCurrentQueueTimeMillis = 1000,
        // 关闭消费者时等待消息消费的最长时间，以毫秒为单位。最小值为 0
        awaitTerminationMillisWhenShutdown = 1000
)
public class DLQSendDataTypeConsumer implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {

    private static final Logger log = LoggerFactory.getLogger(DLQSendDataTypeConsumer.class);

    /**
     * <br>1、RocketMQ-Console在死信队列里面可以重新发送该消息来消费
     * <br>死信队列：没有被及时消费的消息存放的队列，消息没有被及时消费有以下几点原因：
     * <br>       a.消息被拒绝（basic.reject/ basic.nack）并且不再重新投递 requeue=false
     * <br>       b.TTL(time-to-live) 消息超时未消费
     * <br>       c.达到最大队列长度
     * <br>本文从消费者的业务代码出现异常讲起，介绍了 RocketMQ 的重试队列和死信队列：
     * <br>代码正常执行返回消息状态为CONSUME_SUCCESS，执行异常返回RECONSUME_LATER
     * <br>状态为RECONSUME_LATER的消息会进入到重试队列，重试队列的名称为 %RETRY% + ConsumerGroupName；
     * <br>重试16次消息任然没有处理成功，消息就会进入到死信队列%DLQ% + ConsumerGroupName;
     */
    @Override
    public void onMessage(MessageExt messageExt) {
        int reconsumeTimes = messageExt.getReconsumeTimes();
        log.info("死信队列reconsumeTimes:{}", reconsumeTimes);
        MqTestVO mqTestVO = JacksonUtil.toObject(new String(messageExt.getBody()), MqTestVO.class);
        if (Objects.equals("111", mqTestVO.getUserId())) {
            log.info("在死信队列里面消费成功了：{}", JacksonUtil.toJson(mqTestVO));
        }
        log.info(
                String.format("在死信队列里面消费成功了 name is %s data is %s", this.getClass().getSimpleName(), mqTestVO));
    }

    @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {
        log.info("consumer:{}", consumer.toString());
    }
}
